package SSF.OS.NetFlow.Filter;

import com.renesys.raceway.DML.*;
import SSF.Util.Streams.*;
import SSF.OS.NetFlow.*;
import java.util.*;
import java.io.*;
import java.net.*;


// This is MOSTLY copied from SSF.Util.Stream.BasicPlayer !! It's not extended 
// directly from BasicPlayer because of some changes that won't be comfortably
// allowed by java, plus, there might still be some changes for FilterPlayer or 
// even BasicPlayer.
//
// CHANGES ARE (compared with BasicPlayer):
// -- a filter is added.
// -- sourceDictionary, typeDictionary are changed from private to protected
// -- method addType, addSource were added, so that the user would have the option
// to write some of their own code when a record of "type" or "source nhi" is met.
 //

/**
 * It reads records from the net flow data file, pass all the records through 
 * a filter. It's NOT an extension of <I>SSF.Util.Streams.BasicPlayer</I>, but
 * it might be in the future release.
 */
public class FilterPlayer implements StreamInterface
{
    /** filter is not from BasicPlayer */
    protected Filter filter = null;

    protected String streamID;

    /** in BasicPlayer, it's private member */
    protected Hashtable sourceDictionary;

    /** in BasicPlayer, it's private member */
    protected Hashtable typeDictionary;

    private DataInputStream din;

    private boolean connected = false;

    private long pct,bct,ms;

    public FilterPlayer (String streamID)
    {
	typeDictionary = new Hashtable();
	sourceDictionary = new Hashtable();
	this.streamID = streamID;
	pct = bct = ms = 0;
    }
    
    public void assignFilter(Filter pFilter)
    {
	filter = pFilter;
    }
    public void connectWrite(String url) throws streamException {
	throw new streamException("Player cannot connectWrite(\""+url+"\")");
    }

    public void disconnect() {
	if (connected) {
	    try {
		din.close();
		din=null;
	    } catch (Exception any) {
		// ignore
	    }
	}
	connected = false;
    }

    public void connectRead(String url) throws streamException {
	if (connected) 
	    throw new streamException("Cannot connect: already connected");

	if (url.indexOf(":")<0) url = "file:"+url;

	if (url.startsWith("file:")) {
	    try {
		String fname = url.substring(5);
		File f = new File(fname);
		din = new DataInputStream(new FileInputStream(f));
	    } catch (Exception any) {
		throw new streamException("Cannot connect: "+any.toString());
	    }
	} 

	else if (url.startsWith("jahnu:")) { 
	    // Jahnu stream database from Renesys Corporation
	    try {
		URL U = new URL(url);
		String hostname = U.getHost();
		int portnum = U.getPort();
		Socket sock = new Socket(hostname,portnum);
		DataOutputStream req = new DataOutputStream
		    (new BufferedOutputStream(sock.getOutputStream()));
		req.writeUTF("play");
		req.writeUTF(streamID==null?"*":streamID);
	    } catch (Exception any) {
		throw new streamException("Cannot connect: "+any.toString());
	    }
	} 

	else throw new streamException("Cannot connect: malformed URL: "+url);

	try {
	    String cmd = din.readUTF();
	    if (!cmd.equals("record")) 
		throw new streamException("Bad header command: \""+cmd+
					  "\" (expected 'record')");
	    
	    String sid = din.readUTF();
	    if (streamID!=null && !sid.equals(streamID))
		throw new streamException("Stream ID mismatch \""+sid+
					  "\" (expected '"+streamID+"')");
	    
	    connected = true;
	} catch (IOException ioex) {
	    throw new streamException(ioex.toString());
	}

	
	byte[] bytes = new byte[8192];

	ms = System.currentTimeMillis();

	while (true) {
	    try {
		int typecode = din.readInt();
		int source = din.readInt();
		double time = din.readDouble();
		int len = din.readInt();
		if (len>bytes.length) bytes = new byte[len];
		din.read(bytes,0,len);

		switch(typecode) {
		case 0:
		    addType(new String(bytes,0,len));
		    break;
		case 1:
		    addSource(new String(bytes,0,len));
		    break;
		default:
		    receive(typecode,source,time,bytes,0,len);
		    break;
		}
			
		pct++; bct += len;

	    } catch (EOFException eof) {

		connected = false;

		ms = System.currentTimeMillis() - ms;

		System.err.println("{Player processed "+pct+" records, "+
				   bct+" bytes, in "+ms/1000.+" seconds ("+
				   (bct/ms)+" KB/s)}");

		return;

	    } catch (Exception any) {
		throw new streamException(any.toString());
	    }
	}
    }

    public int send(int type_code,
		    int source_code,
		    double timestamp, 
		    byte[] bytes, 
		    int offset,
		    int length) { 
	return send(type_code,source_code,timestamp);
    }

    public int send(int type_code,
		    int source_code,
		    double timestamp) { 
	System.err.println("Can't use FilterPlayer to send()");
	return -1;
    }

    public boolean isConnected() { 
	return connected;
    }

    //-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
    // DICTIONARY SUPPORT

    public String getRecordTypeString(int id) {
	return (String)typeDictionary.get(new Integer(id));
    }

    public int getRecordTypeCode(String name) {
	Integer I = (Integer)typeDictionary.get(name);
	if (I==null) return -1; else return I.intValue();
    }

    public String getRecordSourceString(int id) {
	return (String)sourceDictionary.get(new Integer(id));
    }

    public int getRecordSourceCode(String name) {
	Integer I = (Integer)sourceDictionary.get(name);
	if (I==null) return -1; else return I.intValue();
    }

    ///////////Methods not from BasicPlayer////////////////////////////////////
    /**
     * operations after a record of netflow stream is met
     * if the record is NOT a netflow record, this function should be overloaded.
     */
    public int receive(int type_code, int source_code, double timestamp,
		       byte[] buf, int offset, int length)
    {
	String src;
	String typ;


	FilterData fdata = 
	    new FilterData(type_code, source_code, timestamp, buf);

	if (filter==null || filter.passFilter(fdata)) {
	    src = getRecordSourceString(source_code);
	    if (src==null) src="?unknown source";
	    typ = getRecordTypeString(type_code);
	    if (typ==null) typ="?unknown type";

	    System.out.println("[type="+type_code+" (\""+typ+"\") "+
			       " source="+source_code+" (\""+src+"\") "+
			       " time="+timestamp+" "+
			       " bytes="+(length-offset)+"]");
	    //for test when it's "SSF.OS.NetFlow"
	    //IpNetFlow flow = new IpNetFlow();
	    //flow.readFromBytes(buf, length);
	    //System.out.println(flow.toString());
	}

	//for debug
	//System.err.println("Finish filtering.");

	return 0;    //success	
    }
    /**
     * The method that will be called when a record of "type" announcement is 
     * met in the inputStream.
     * @param spec the string generated from the record.
     */
    protected void addType(String spec) 
	throws streamException
    {
	int sp = spec.indexOf(" ");
	if (sp > 0) {
	    Integer I = new Integer(spec.substring(0,sp));
	    String N = spec.substring(sp+1).trim();
	    typeDictionary.put(I,N);
	    typeDictionary.put(N,I);

	    //put in decoder for the filter
	    //if this is not a record type, no decoder will be added.
	    if (filter != null)
	        filter.addDecoder(N, I);
	} 
	else throw new streamException("Bad code: "+spec);
    }

    /**
     * The method that will be called when a record of "source nhi" announcement is
     * met in the inputStream.
     * @param spec the string generated from the record.
     */
    protected void addSource(String spec)
	throws streamException
    {
	int sp = spec.indexOf(" ");
	if (sp > 0) {
	    Integer I = new Integer(spec.substring(0,sp));
	    String N = spec.substring(sp+1).trim();
	    sourceDictionary.put(I,N);
	    sourceDictionary.put(N,I);
	}
	else throw new streamException("Bad code: "+spec);
    }

    ///////////////////////////////main//////////////////////////////////////////
    /**
     * usage: java FilterPlayer &lt;filter config> &lt;flow data file> &lt;stream name>
     */
    public static void main(String[] args) {

	//check arguments
	if (args.length < 3) {
	    System.err.println("usage: java FilterPlayer <filter configfile> <flow data file> <stream name>");
	    return;
	}
	
	//create FilterPlayer
	FilterPlayer fPlayer = new FilterPlayer(args[2]);
	Filter pFilter = null;

	//config filter
	dmlConfig filterConfig = null;

	filterConfig = new dmlConfig(args[0]);

	Configuration fConfig = (Configuration)filterConfig.findSingle(".filter");
	if (fConfig == null) {
	    System.err.println("Configuration of filter must define a .filter attribute");
	    System.err.println(fConfig);
	    System.exit(-1);
	}

	try {
	    pFilter = new Filter(fPlayer, fConfig);
	}
	catch (configException cfe) {
	    System.err.println(cfe.getMessage());
	    System.exit(-1);
	}
	
	//for debug
	System.err.println("Complete configuring the fiter");

	fPlayer.assignFilter(pFilter);

	try {
	    fPlayer.connectRead(args[1]);
	}
	catch (streamException se) {
	    System.err.println(se.getMessage());
	}
    }
}
